package com.ynding.spring.physical.redis.meta.queue.delayed;

import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RBlockingQueue;
import org.redisson.api.RedissonClient;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
import org.springframework.stereotype.Component;

import java.util.Map;

/**
 * <p> 队列初始化 </p>
 *
 * @author ynding
 * @version 2022/4/20
 **/
@Component
@RequiredArgsConstructor
@Slf4j
public class RedisDelayedQueueInit implements ApplicationContextAware {
    private final RedissonClient redissonClient;
    private final ThreadPoolTaskScheduler threadPoolTaskScheduler;

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        Map<String, RedisDelayedQueueListener> map = applicationContext.getBeansOfType(RedisDelayedQueueListener.class);
        for (Map.Entry<String, RedisDelayedQueueListener> taskEventListenerEntry : map.entrySet()) {
            String listenerName = taskEventListenerEntry.getValue().getClass().getName();
            startThread(listenerName, taskEventListenerEntry.getValue());
        }
    }

    /**
     * 启动线程获取队列*
     *
     * @param queueName                 queueName
     * @param redisDelayedQueueListener 任务回调监听
     * @param <T>                       泛型
     * @return
     */
    private <T> void startThread(String queueName, RedisDelayedQueueListener redisDelayedQueueListener) {
        RBlockingQueue<T> blockingFairQueue = redissonClient.getBlockingQueue(queueName);
        //服务重启后，无offer，take不到信息。
        redissonClient.getDelayedQueue(blockingFairQueue);
        log.info("启动监听队列线程" + queueName);
        threadPoolTaskScheduler.scheduleWithFixedDelay(() -> {
            try {
                // 使用take 直至消息到达，否则等待
                T t = blockingFairQueue.take();
                log.info("监听队列线程，监听名称：{},内容:{}", queueName, t);
                redisDelayedQueueListener.invoke(t);
            } catch (Exception e) {
                log.info("监听队列线程错误,", e);
            }
        }, 2000);
    }
}
